package com.abyss.state;


import com.google.inject.internal.util.$SourceProvider;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 做一个单词统计的demo
 * 使用reduce方法 来完成聚合同时做状态的记录
 */
public class KeyedStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 使用state必须开启checkpoint
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new FsStateBackend("file:///G:\\26-code\\Java\\flink_base_sh\\data\\output\\checkpoint"));

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordWithNum = env.socketTextStream("node1", 9999).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] ss = value.split(" ");
                return Tuple2.of(ss[0], Integer.parseInt(ss[1]));
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordWithNum.keyBy(0).reduce(new RichReduceFunction<Tuple2<String, Integer>>() {
            private ValueState<Integer> state;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                state = getRuntimeContext().getState(new ValueStateDescriptor<Integer>(
                        "test value state",
                        TypeInformation.of(new TypeHint<Integer>() {
                        })
                ));
            }

            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                if (state.value() == null) {
                    int accumulator = value1.f1 + value2.f1;
                    Tuple2<String, Integer> result = Tuple2.of(value1.f0, accumulator);

                    state.clear();  // 清空过期的数据
                    state.update(accumulator);
                    return result;
                }else {
                    int accumulator = state.value() + value2.f1;
                    Tuple2<String, Integer> result = Tuple2.of(value1.f0, accumulator);

                    state.clear();  // 清空过期的数据
                    state.update(accumulator);
                    return result;
                }
            }
        });

        result.print();
        env.execute();
    }
}